package cn.jly.flink.source2sink.kafka;

import cn.jly.flink.entity.Metric;
import cn.jly.flink.utils.ConstantUtils;
import cn.jly.flink.utils.FlinkUtils;
import com.alibaba.fastjson.JSON;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import org.apache.flink.util.Collector;
import org.apache.kafka.clients.producer.ProducerConfig;

import java.io.File;
import java.nio.charset.StandardCharsets;
import java.util.Properties;

/**
 * @PackageName cn.jly.flink.source
 * @ClassName KafkaSinkDemo
 * @Description
 * @Author 姬岚洋
 * @Date 2021/1/13 下午 5:08
 */
public class KafkaSinkDemo {
    public static void main(String[] args) throws Exception {
        final StreamExecutionEnvironment env = FlinkUtils.getStreamExecutionEnv();
        final File file = new File("D:\\test\\2021-01-14--20\\.part-11-0.inprogress.4a6f25a7-f594-43a9-850d-b0756af11489");

        final Properties properties = new Properties();
        properties.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, ConstantUtils.Kafka.BOOTSTRAP_SERVERS_VALUE);

        env.readTextFile(file.getAbsolutePath(), StandardCharsets.UTF_8.name())
                .flatMap(new FlatMapFunction<String, Metric>() {
                    @Override
                    public void flatMap(String s, Collector<Metric> collector) throws Exception {
                        if (StringUtils.isNotEmpty(s)) {
                            final Metric metric = JSON.parseObject(s, Metric.class);
                            collector.collect(metric);
                        }
                    }
                })
                .map(new MapFunction<Metric, String>() {
                    @Override
                    public String map(Metric metric) throws Exception {
                        return JSON.toJSONString(metric);
                    }
                })
                .addSink(new FlinkKafkaProducer<>(
                        "metric_test",
                        new SimpleStringSchema(),
                        properties
                ));

        FlinkUtils.executeStream(env, "KafkaSinkDemo");
    }
}
